package blue.mqtt4j.internal.client.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Jin Zheng
 * @since 2020-10-18
 */
public class TaskExecutor
{
	private static Logger logger = LoggerFactory.getLogger(TaskExecutor.class);
	private static final int MIN_THREAD = 10;
	private static final int KEEP_ALIVE = 60;
	private static final int ZERO = 0;

	private static ExecutorService executor;
	private static AtomicInteger counter;
	private final int maxThreads;

	public TaskExecutor(int maxThreads)
	{
		this.maxThreads = maxThreads;
		this.init();
		counter.incrementAndGet();
		logger.debug("Increase counter: {}", counter.get());
	}

	private void init()
	{
		if (executor == null)
		{
			synchronized (this)
			{
				if (executor == null)
				{
					executor = new ThreadPoolExecutor(MIN_THREAD, maxThreads, KEEP_ALIVE, TimeUnit.SECONDS,
							new SynchronousQueue<>());
					counter = new AtomicInteger(ZERO);
					logger.info("Initialize TaskExecutor, max_thread: {}", maxThreads);
				}
			}
		}
	}

	public void execute(Runnable runnable)
	{
		executor.execute(runnable);
	}

	public void shutdown()
	{
		int count = counter.decrementAndGet();
		logger.debug("Decrease counter: {}", counter.get());
		if (executor != null && count <= 0)
		{
			executor.shutdown();
			executor = null;
			logger.info("Shutdown TaskExecutor.");
		}
	}

	public ExecutorService getExecutor()
	{
		return executor;
	}

	public int getMaxThreads()
	{
		return maxThreads;
	}
}
